package com.atguigu.gmall.realtime.app.dwd.db;

import com.atguigu.gmall.realtime.app.BaseSQLApp;
import com.atguigu.gmall.realtime.common.Constant;
import com.atguigu.gmall.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @Author lzc
 * @Date 2022/4/30 10:34
 */
public class DWD_05_DwdTradeOrderPreApp extends BaseSQLApp {
    public static void main(String[] args) {
        new DWD_05_DwdTradeOrderPreApp().init("DwdTradeOrderPreApp", 3006, 1, "DwdTradeOrderPreApp");
    }
    
    @Override
    public void handle(StreamExecutionEnvironment env,
                       StreamTableEnvironment tEnv) {
        
        // TODO 5张表join:
        //      订单明细  订单  订单明细活动关联表  订单明细优惠券关联表  字典表
        
        // 因为有join, 锁状态保留时间
        tEnv.getConfig().setIdleStateRetention(Duration.ofDays(1));
        // 1. 建表从kafka读取数据 ods_db
        tEnv.executeSql("create table ods_db(" +
                            "   `database` string, " +
                            "   `table` string, " +
                            "   `type` string, " +
                            "   `ts` bigint, " +
                            "   `data` map<String, String>, " +
                            "   `old` map<String, String>," +
                            "   `pt` as proctime()  " +  // 将来lookup join 的时候使用
                            ")" + SQLUtil.getKafkaDDL("DwdTradeOrderPreApp", Constant.TOPIC_ODS_DB));
        // 2. 读取订单明细数据
        Table orderDetail = tEnv
            .sqlQuery("select  " +
                          "data['id'] id, " +  // 订单明细id
                          "data['order_id'] order_id, " +  // 订单id
                          "data['sku_id'] sku_id, " +  // sku_id
                          "data['sku_name'] sku_name, " +
                          "data['create_time'] create_time, " +
                          "data['source_id'] source_id, " +  //
                          "data['source_type'] source_type, " +
                          "data['sku_num'] sku_num, " +
                          "cast(cast(data['sku_num'] as decimal(16,2)) * " +
                          "cast(data['order_price'] as decimal(16,2)) as String) split_original_amount, " +  // 分摊原始总金额, 就是没有 -优惠券-活动+运费
                          "data['split_total_amount'] split_total_amount, " +  // 分摊总金额, 已经 -优惠券-活动+运费
                          "data['split_activity_amount'] split_activity_amount, " +  // 分摊活动金额
                          "data['split_coupon_amount'] split_coupon_amount, " + // 分摊优惠券金额
                          "ts od_ts, " +   // 时间戳
                          "pt " +  // 处理时间
                          "from `ods_db` where `table` = 'order_detail' " +   // 过滤出订单明细表数据
                          "and `type` = 'insert' "); // 订单明细, 只需要新增数据就行了
        tEnv.createTemporaryView("order_detail", orderDetail);
        
        //3. 读取订单表数据 order_info
        Table orderInfo = tEnv
            .sqlQuery("select  " +
                          "data['id'] id, " +  //
                          "data['user_id'] user_id, " +
                          "data['province_id'] province_id, " +
                          "data['operate_time'] operate_time, " +
                          "data['order_status'] order_status, " +  // 订单状态
                          "`type`, " +  // 将来和订单明细表进行join的时候需要区别insert和update
                          "`old`, " +  // 因为会用到update相关信息, 需要把old数据取出来
                          "ts oi_ts " +  // 订单表时间戳
                          "from `ods_db` " +
                          "where `table` = 'order_info' " +
                          "and (`type` = 'insert' or `type` = 'update')");
        tEnv.createTemporaryView("order_info", orderInfo);
        
        
        // 4. 读取订单明细活动关联表数据
        Table orderDetailActivity = tEnv
            .sqlQuery("select  " +
                          "data['order_detail_id'] order_detail_id, " +  //
                          "data['activity_id'] activity_id, " +
                          "data['activity_rule_id'] activity_rule_id " +
                          "from `ods_db` " +
                          "where `table` = 'order_detail_activity' " +
                          "and `type` = 'insert' ");
        tEnv.createTemporaryView("order_detail_activity", orderDetailActivity);
        // 5. 读取订单明细优惠券关联表数据
        Table orderDetailCoupon = tEnv
            .sqlQuery("select " +
                          "data['order_detail_id'] order_detail_id, " +
                          "data['coupon_id'] coupon_id " +
                          "from `ods_db` " +
                          "where `table` = 'order_detail_coupon' " +
                          "and `type` = 'insert' ");
        tEnv.createTemporaryView("order_detail_coupon", orderDetailCoupon);
        
        //6. 建立 MySQL-LookUp 字典表
        tEnv.executeSql(SQLUtil.getBaseDicLookUpDDL());
        
        
        // 7. 关联五张表获得订单明细表
        Table resultTable = tEnv
            .sqlQuery("select  " +
                          "od.id, " +
                          "od.order_id, " +
                          "oi.user_id, " +
                          "oi.order_status, " +
                          "od.sku_id, " +
                          "od.sku_name, " +
                          "oi.province_id, " +
                          "act.activity_id, " +
                          "act.activity_rule_id, " +
                          "cou.coupon_id, " +
                          "date_format(od.create_time, 'yyyy-MM-dd') date_id, " +
                          "od.create_time, " +
                          "date_format(oi.operate_time, 'yyyy-MM-dd') operate_date_id, " +
                          "oi.operate_time, " +
                          "od.source_id, " +
                          "od.source_type, " +
                          "dic.dic_name source_type_name, " +
                          "od.sku_num, " +
                          "od.split_original_amount, " +
                          "od.split_activity_amount, " +
                          "od.split_coupon_amount, " +
                          "od.split_total_amount, " +
                          "oi.`type`, " +
                          "oi.`old`, " +
                          "od.od_ts, " +
                          "oi.oi_ts, " +
                          "current_row_timestamp() row_op_ts " +
                          "from order_detail od  " +
                          "join order_info oi " +  // 订单和详情用 内连接
                          "on od.order_id = oi.id " +
                          "left join order_detail_activity act " + // 有些订单没有活动, 所以使用left join
                          "on od.id = act.order_detail_id " +
                          "left join order_detail_coupon cou " + // 有些订单没有使用优惠券所以使用 left join
                          "on od.id = cou.order_detail_id " +
                          "join `base_dic` for system_time as of od.pt as dic " + // 连接字典表的时候也用内连接
                          "on od.source_type = dic.id");
        tEnv.createTemporaryView("result_table", resultTable);
        
        
        // 8. 建立 Upsert-Kafka dwd_trade_order_pre_process 表
        tEnv.executeSql("create table dwd_trade_order_pre_process( " +
                            "id string, " +
                            "order_id string, " +
                            "user_id string, " +
                            "order_status string, " +
                            "sku_id string, " +
                            "sku_name string, " +
                            "province_id string, " +
                            "activity_id string, " +
                            "activity_rule_id string, " +
                            "coupon_id string, " +
                            "date_id string, " +
                            "create_time string, " +
                            "operate_date_id string, " +
                            "operate_time string, " +
                            "source_id string, " +
                            "source_type string, " +
                            "source_type_name string, " +
                            "sku_num string, " +
                            "split_original_amount string, " +
                            "split_activity_amount string, " +
                            "split_coupon_amount string, " +
                            "split_total_amount string, " +
                            "`type` string, " +
                            "`old` map<string,string>, " +
                            "od_ts bigint, " +
                            "oi_ts bigint, " +
                            "row_op_ts timestamp_ltz(3), " +
                            "primary key(id) not enforced " +
                            ")" + SQLUtil.getUpsertKafkaDDL("dwd_trade_order_pre_process"));
        
        resultTable.executeInsert("dwd_trade_order_pre_process");
        
        
    }
}
/*
 
 
 */